package hasthi.container.p2p;

import hasthi.common.BaseContext;
import hasthi.common.HasthiError;
import hasthi.common.HasthiException;
import hasthi.common.MessageRendezvous;
import hasthi.common.Utils;
import hasthi.common.XmlBeanUtils;
import hasthi.common.HasthiException.FaultCode;
import hasthi.common.constants.HasthiConstants;
import hasthi.container.CommunicationClient;
import hasthi.container.CommunicationContianier;
import hasthi.container.ContainerUtils;
import hasthi.container.MessageContext;
import hasthi.container.MessageProcessor;
import hasthi.container.NodeAddress;
import hasthi.container.SOAPUtils;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.UnknownHostException;
import java.util.Collection;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.xmlbeans.XmlObject;
import org.w3.x2005.x08.addressing.RelatesToDocument;
import org.w3.x2005.x08.addressing.ToDocument;

import rice.environment.Environment;
import rice.p2p.commonapi.Application;
import rice.p2p.commonapi.Endpoint;
import rice.p2p.commonapi.Id;
import rice.p2p.commonapi.Message;
import rice.p2p.commonapi.RouteMessage;
import rice.p2p.scribe.ScribeClient;
import rice.p2p.scribe.ScribeContent;
import rice.p2p.scribe.ScribeImpl;
import rice.p2p.scribe.Topic;
import rice.pastry.NodeHandle;
import rice.pastry.NodeIdFactory;
import rice.pastry.PastryNode;
import rice.pastry.PastryNodeFactory;
import rice.pastry.commonapi.PastryIdFactory;
import rice.pastry.leafset.LeafSet;
import rice.pastry.routing.RoutingTable;
import rice.pastry.socket.SocketPastryNodeFactory;
import rice.pastry.standard.RandomNodeIdFactory;

/**
 * Implementation of a Peer to Peer Container,, implemented with FreePastry. It
 * uses a binery protocol. However we send SOAP message for all interactions. So
 * it will be like using JMX for SOAP.
 * 
 * @author Srinath Perera(hperera@cs.indiana.edu)
 */

public class P2PContainer implements Application, ScribeClient, CommunicationContianier, MulticastContianier {
    private static Log log = LogFactory.getLog(HasthiConstants.LOGGER_NAME);
    private MessageProcessor messageProcessor;
    private PastryNode node;
    /**
     * The Endpoint represents the underlieing node. By making calls on the
     * Endpoint, it assures that the message will be delivered to a MyApp on
     * whichever node the message is intended for.
     */
    protected Endpoint endpoint;
    protected ScribeImpl myScribe;
    protected Topic myTopic;

    private int bindport;
    private InetSocketAddress bootaddress;
    private Environment env;
    private Map<String, BroadcastResponseBag> broadcastMap = new Hashtable<String, BroadcastResponseBag>();
    private int peerCount;
    private NodeIdFactory nidFactory;

    private Broadcaster broadcaster;
    private boolean useScribe = false;

    /**
     * this is a syncronizing point for requests and responses.
     */
    protected MessageRendezvous messageRendezvous = new MessageRendezvous();

//    public P2PContainer(MessageProcessor managerInstance) throws HasthiException {
//        this(managerInstance, "linbox3.extreme.indiana.edu", 9001);
//    }
//
//    public P2PContainer(MessageProcessor managerInstance, int bindport) throws HasthiException {
//        this(managerInstance, "linbox3.extreme.indiana.edu", 9001, bindport);
//    }

//    public P2PContainer(MessageProcessor managerInstance, String bootStrapHost, int bootport)
//            throws HasthiException {
//        this(managerInstance, bootStrapHost, bootport, -1);
//    }

    public P2PContainer(BaseContext baseContext, MessageProcessor managerInstance) throws HasthiException{
        this(baseContext,managerInstance,baseContext.getBoostrapHost(),baseContext.getBoostrapP2PPort(),-1);
    }
    
    public P2PContainer(BaseContext baseContext, MessageProcessor managerInstance, String bootStrapHost, int bootport) throws HasthiException{
        this(baseContext,managerInstance,bootStrapHost,bootport,-1);
    }

    
    public P2PContainer(BaseContext baseContext, MessageProcessor managerInstance,
            String bootStrapHost, int bootport,int bindport) throws HasthiException {
        try {
            this.messageProcessor = managerInstance;
            // Loads pastry settings
            env = new Environment();
            // disable the UPnP setting (in case you are testing this on a
            // NATted
            // LAN)
            env.getParameters().setString("nat_search_policy", "never");

            if (bindport == -1) {
                ServerSocket server = new ServerSocket(0);
                server.close();
                this.bindport = server.getLocalPort();
            } else {
                this.bindport = bindport;
            }

            // build the bootaddress from the command line args
            InetAddress bootaddr = InetAddress.getByName(bootStrapHost);
            bootaddress = new InetSocketAddress(bootaddr, bootport);
            broadcaster = new Broadcaster(this);
        } catch (UnknownHostException e) {
            throw new HasthiException(e, FaultCode.InitalizationError);
        } catch (IOException e) {
            throw new HasthiException(e, FaultCode.InitalizationError);
        }
    }

    /* (non-Javadoc)
     * @see hasthi.container.p2p.MulticastContianier#startContainer()
     */
    public void startContainer() throws HasthiException {
        try {
            // Generate the NodeIds Randomly
            nidFactory = new RandomNodeIdFactory(env);

            // construct the PastryNodeFactory, this is how we use
            // rice.pastry.socket
            PastryNodeFactory factory = new SocketPastryNodeFactory(nidFactory, bindport, env);

            // This will return null if we there is no node at that location
            NodeHandle bootHandle = ((SocketPastryNodeFactory) factory).getNodeHandle(bootaddress);

            // construct a node, passing the null boothandle on the first loop
            // will
            // cause the node to start its own ring
            node = factory.newNode(bootHandle);

            // the node may require sending several messages to fully boot into
            // the
            // ring
            long starttime = System.currentTimeMillis();
            synchronized (node) {
                while (!node.isReady() && !node.joinFailed()) {
                    if((System.currentTimeMillis() - starttime) > 5*60*1000 ){
                        throw new HasthiException("Took too long to Join the managment Ring "
                                + node.joinFailedReason(),FaultCode.InitalizationError);
                    }
                    // delay so we don't busy-wait
                    node.wait(500);

                    // abort if can't join
                    if (node.joinFailed()) {
                        throw new HasthiException("Could not join the FreePastry ring.  Reason:"
                                + node.joinFailedReason(),FaultCode.InitalizationError);
                    }
                }
            }
            // We are only going to use one instance of this application on each
            // PastryNode
            this.endpoint = node.buildEndpoint(this, "myinstance");
            // the rest of the initialization code could go here
            this.endpoint.setDeserializer(new XmlMessageDeserializer());
            myScribe = new ScribeImpl(node, "myScribeInstance");
            // construct the topic
            myTopic = new Topic(new PastryIdFactory(node.getEnvironment()), "admin");
            // now we can receive messages
            this.endpoint.register();
            myScribe.subscribe(myTopic, this);

            System.out.println("Node started as " + node.getId());

//            ManagerContext managerContext = managerInstance.getManagerContext();
//            managerContext.setP2pContianer(this);
            // TODO remove
            // ManagerAddress address = managerContext.getMyAddress();
            // if(address == null){
            // address = ManagerAddress.Factory.newInstance();
            // managerContext.setMyAddress(address);
            // }else{
            // address.setP2Paddress(node.getId().toStringFull());
            // }
            env.getTimeSource().sleep(1000);
            if(messageProcessor != null){
                messageProcessor.addContainer(this);    
            }
        } catch (IOException e) {
            throw new HasthiException(e, FaultCode.InitalizationError);
        } catch (InterruptedException e) {
            throw new HasthiException(e, FaultCode.InitalizationError);
        }
    }

    /* (non-Javadoc)
     * @see hasthi.container.p2p.MulticastContianier#stopContainer()
     */
    public void stopContainer() throws HasthiException {
        node.destroy();
        System.out.println("p2p Node " + getNode() + " stopped");
    }

    public XmlMesssage checkForReplies(List<String> ids, MessageFilter filter) {
        XmlMesssage response;
        for (String id : ids) {
            response = messageRendezvous.getMessage(id);
            if (response != null && filter.usefulMessage(response)) {
                return response;
            }
        }
        log.debug("[" + node.getId() + "]No replies found");
        return null;
    }

    /* (non-Javadoc)
     * @see hasthi.container.p2p.MulticastContianier#anyCastAndReceive(hasthi.container.MessageContext)
     */
    public XmlObject anyCastAndReceive(MessageContext messageContext) throws HasthiException {
        UnicastXmlMessage requestMessage = new UnicastXmlMessage(messageContext.getRequest(), node
                .getId(), messageContext.getSoapAction());
        return anyCastAndReceive(requestMessage);
    }

    private XmlObject anyCastAndReceive(UnicastXmlMessage message) throws HasthiException {
        try {
            LeafSet leafSet = node.getLeafSet();
            Id targetId = null;
            
            // If a leaf node has a value, we send message to them
            int startIndex = Math.abs((int) (System.currentTimeMillis() % leafSet.size()));
            for(int i = 0;i<leafSet.size();i++){
                NodeHandle handle = leafSet.get(Math.abs((startIndex+i)%leafSet.size()));
                // if entry it myself, it does no good to us
                if (handle != null && !node.getId().equals(handle.getId())) {
                    targetId = handle.getId();
                    break;
                }
            }
            
            if (targetId == null) {
                // otherwise we send a message to first entry in routing table

                // for(int i = 0;i<rt.numRows();i++){
                // RouteSet[] rows = rt.getRow(i);
                // for(RouteSet set:rows){
                // if(set != null && set.closestNode() != null){
                // targetId = set.closestNode().getId();
                // }
                // }
                // }
                /**
                 * We will randomly pick a entry in the routing table
                 */
                RoutingTable rt = node.getRoutingTable();
                List entries = rt.asList();
                startIndex = Math.abs((int) (System.currentTimeMillis() % entries.size()));
                for(int i = 0;i<entries.size();i++) {
                    NodeHandle handle = (NodeHandle) entries
                            .get(Math.abs((startIndex +i)%entries.size()));
                    if (handle != null && !node.getId().equals(handle.getId())) {
                        targetId = handle.getId();
                        break;
                    }
                }
            }
            if (targetId != null) {
                UnicastXmlMessage responseMessage = sendReciveMessage(targetId, message, 10);
                return responseMessage == null ? null : responseMessage.getBodyContent();
            } else {
                return null;
            }
        } catch (InterruptedException e) {
            throw new HasthiException(e, FaultCode.LocalError);
        }
    }

    public void anyCast(MulticastXmlMessage anycastMessage) throws HasthiException,
            InterruptedException {
        myScribe.anycast(myTopic, anycastMessage);
    }

    public UnicastXmlMessage anyCastAndWait(XmlObject message, String soapAction)
            throws HasthiException, InterruptedException {
        MulticastXmlMessage anycastMessage = new MulticastXmlMessage(message, node.getId(),
                soapAction);
        myScribe.anycast(myTopic, anycastMessage);

        log.debug("Waiting for " + endpoint.getId() + " respond ");
        String reqeustMessageID = (String) anycastMessage.getHeader(ContainerUtils.WSA_MESSAGEID);
        // Wait for 30 seconds
        return (UnicastXmlMessage) messageRendezvous.waitFor(reqeustMessageID, 30);
    }

    /** methods for mulicasting * */
    public void mulitiCast(XmlObject message, String soapAction) {
        MulticastXmlMessage multiCastMessage = new MulticastXmlMessage(message, node.getId(),
                soapAction);
        myScribe.publish(myTopic, multiCastMessage);
    }

    public Collection<XmlObject> mulitiCastAndWait(XmlObject message, String soapAction,
            long waitTimeInMs, int expectedResponseCount) throws InterruptedException {
        log.debug("Multiciating from " + getNode());
        boolean expectResponses = expectedResponseCount > 0;
        XmlMesssage multiCastMessage;

        if (useScribe) {
            multiCastMessage = new MulticastXmlMessage(message, node.getId(), soapAction);
        } else {
            multiCastMessage = new UnicastXmlMessage(message, node.getId(), soapAction);
        }
        String messageID = multiCastMessage.getMessageID();
        if(expectResponses){
            broadcastMap.put(messageID, new BroadcastResponseBag());    
        }
        if (useScribe) {
            myScribe.publish(myTopic, (MulticastXmlMessage) multiCastMessage);
        } else {
            broadcaster.initiateBroadcast((UnicastXmlMessage) multiCastMessage);
        }

        
        if(expectResponses){
            // Wait for responses
            int sleptTime = 0;
            while (sleptTime < waitTimeInMs) {
                env.getTimeSource().sleep(1000);
                sleptTime = sleptTime + 1000;
                if (broadcastMap.get(messageID).size() >= expectedResponseCount) {
                    break;
                }
            }

            // List<XmlMessage> responseMessages =
            // broadcastMap.remove(messageID).getMessages();
            // List<XmlObject> results = new
            // ArrayList<XmlObject>(responseMessages.size());
            // for(XmlMessage responseMsg:responseMessages){
            // try {
            // results.add(responseMsg.getBodyContent());
            // } catch (HasthiException e) {
            // //this should not happen, we throw a error so we can see the error
            // and fix it if it happen
            // throw new HasthiError(e);
            // }
            // }
            // log.info(results.size() +"/" + expectedResponseCount + " responses
            // recived");
            // return results;

            Map<String, XmlObject> results = new HashMap<String, XmlObject>();
            List<UnicastXmlMessage> responseMessages = broadcastMap.remove(messageID).getMessages();

            for (UnicastXmlMessage responseMsg : responseMessages) {
                try {
                    if (!results.containsKey(responseMsg.getReplyTo())) {
                        results.put(responseMsg.getReplyTo(), responseMsg.getBodyContent());
                    }
                } catch (HasthiException e) {
                    // this should not happen, we throw a error so we can see the
                    // error and fix it if it happen
                    throw new HasthiError(e);
                }
            }
            log.debug(results.size() + "/" + expectedResponseCount + " responses recived");
            return results.values();
        }else{
          //if user dones not expect any responses, just return
            return null;
        }
    }

    /* (non-Javadoc)
     * @see hasthi.container.p2p.MulticastContianier#broadcastAndWait(org.apache.xmlbeans.XmlObject, java.lang.String, long, int)
     */
    public Collection<XmlObject> broadcastAndWait(XmlObject message, String soapAction,
            long waitTimeInMs, int expectedResponseCount) throws InterruptedException {
        log.info("Multiciating from " + getNode());
        UnicastXmlMessage multiCastMessage = new UnicastXmlMessage(message, node.getId(),
                soapAction);
        String messageID = multiCastMessage.getMessageID();
        broadcastMap.put(messageID, new BroadcastResponseBag());

        broadcaster.initiateBroadcast(multiCastMessage);

        // Wait for responses
        int sleptTime = 0;
        while (sleptTime < waitTimeInMs) {
            env.getTimeSource().sleep(1000);
            sleptTime = sleptTime + 1000;
            if (broadcastMap.get(messageID).size() >= expectedResponseCount) {
                break;
            }
        }

        Map<String, XmlObject> results = new HashMap<String, XmlObject>();
        List<UnicastXmlMessage> responseMessages = broadcastMap.remove(messageID).getMessages();

        for (UnicastXmlMessage responseMsg : responseMessages) {
            try {
                if (!results.containsKey(responseMsg.getReplyTo())) {
                    results.put(responseMsg.getReplyTo(), responseMsg.getBodyContent());
                }
            } catch (HasthiException e) {
                // this should not happen, we throw a error so we can see the
                // error and fix it if it happen
                throw new HasthiError(e);
            }
        }
        log.info(results.size() + "/" + expectedResponseCount + " responses recived");
        return results.values();
    }

    /**
     * This method is invoked when a message is delivered for a topic this
     * client is interested in.
     * 
     * @param topic
     *                The topic the message was published to
     * @param content
     *                The content which was published
     */
    public void deliver(Topic topic, ScribeContent message) {
        log.debug("Recive a Mulicast message ");
        if (message instanceof MulticastXmlMessage) {
            processMessage((MulticastXmlMessage) message);
        } else {
            throw new HasthiError("Unknown Message type " + message.getClass());
        }
    }

    /**
     * This method is invoked when an anycast is received for a topic which this
     * client is interested in. The client should return whether or not the
     * anycast should continue.
     * 
     * @param topic
     *                The topic the message was anycasted to
     * @param content
     *                The content which was anycasted
     * @return Whether or not the anycast should continue
     */
    public boolean anycast(Topic topic, ScribeContent message) {
        log.debug("Recived a Anycast");
        if (message instanceof XmlMesssage) {
            return processMessage((XmlMesssage) message);
        } else {
            throw new HasthiError("Unknown Message type " + message.getClass());
        }
    }

    /**
     * Informs this client that a child was added to a topic in which it was
     * interested in.
     * 
     * @param topic
     *                The topic to unsubscribe from
     * @param child
     *                The child that was added
     */
    public void childAdded(Topic tpoic, rice.p2p.commonapi.NodeHandle child) {
        peerCount++;
    }

    /**
     * Informs this client that a child was removed from a topic in which it was
     * interested in.
     * 
     * @param topic
     *                The topic to unsubscribe from
     * @param child
     *                The child that was removed
     */
    public void childRemoved(Topic topic, rice.p2p.commonapi.NodeHandle child) {
        peerCount--;
    }

    /**
     * Informs the client that a subscribe on the given topic failed - the
     * client should retry the subscribe or take appropriate action.
     * 
     * @param topic
     *                The topic which the subscribe failed on
     */
    public void subscribeFailed(Topic topic) {
        try {
            log.info("Subcription falied for " + node.getId());
            env.getTimeSource().sleep(3000);
            myScribe.subscribe(myTopic, this);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    /**
     * Following methods are use for single message routing
     */
    public void deliver(Id id, Message message) {
        if (message instanceof UnicastXmlMessage) {
            processMessage((UnicastXmlMessage) message);
        } else {
            throw new HasthiError("Unknown Message type " + message.getClass());
        }
    }

    /**
     * Process a incomming message by handing it over to underline Message
     * Processor. If there is a return value, it shall be sent using reply to
     * address. If it is a reply to a request associated thread is informed via
     * message messageRendezvous
     * 
     * @param messsage
     * @throws HasthiException
     */
    public boolean processMessage(XmlMesssage messsage) {
        XmlMesssage requestMessage = (XmlMesssage) messsage;
        String soapAction = (String) requestMessage.getAction();
        String replyToAddress = requestMessage.getReplyTo();
        String relatesToID = requestMessage.getRelatesTo();
        String toAddress = requestMessage.getTo();
        
        log.debug("Recived message " + replyToAddress + " -> " + getNode().getId());

        // System.out.print(node.getRoutingTable());
        if (toAddress != null && toAddress.equals(HasthiConstants.BROADCAST_ADDRESS)) {
            broadcaster.broadcast((UnicastXmlMessage) messsage);
        }

        // TODO fix this Ignore the self message, this happens at broadcast
        if (node.getId().toStringFull().equals(replyToAddress)) {
            log.debug("Ignoring self message from " + replyToAddress);
            // log.debug(messsage);
            return false;
        }
        // log.debug("Proceeding with Message " + messsage);
        if (relatesToID != null) {
            BroadcastResponseBag responseList = broadcastMap.get(relatesToID);
            if (responseList != null) {
                // this is a response to a boardcast message sent
                responseList.addMessage((UnicastXmlMessage) messsage);
            } else {
                // this is a reponse to a request already sent
                messageRendezvous.addMessage(relatesToID, requestMessage);
            }
        } else {
            // this is a original request
            try {
                P2PMessageContext messageContext = new P2PMessageContext(requestMessage,new NodeAddress((Id)null,requestMessage.getTo()));
                XmlObject result = messageProcessor.processMessage(messageContext);
                if(log.isTraceEnabled() && result != null){
                    log.trace("Sent:" +XmlBeanUtils.prettyPrint(result));
                }
                if (result != null && replyToAddress != null) {
                    rice.pastry.Id returnId = rice.pastry.Id.build(replyToAddress);
                    UnicastXmlMessage responseMessage = new UnicastXmlMessage(result, node.getId(),
                            soapAction + "Response");
                    RelatesToDocument relatesToDocument = RelatesToDocument.Factory.newInstance();
                    relatesToDocument.addNewRelatesTo().setStringValue(
                            requestMessage.getMessageID());
                    responseMessage.addHeader(relatesToDocument);
                    sendMessage(returnId, responseMessage);
                }
            } catch (Throwable e) {
                try {
                    log.error(e);
                    if (replyToAddress != null) {
                        rice.pastry.Id returnId = rice.pastry.Id.build(replyToAddress);
                        HasthiException fault;
                        if(e instanceof HasthiException){
                            fault = (HasthiException)e;
                        }else{
                            fault = new HasthiException(e,FaultCode.InternalServiceError);
                        }
                        XmlObject faultMessage = fault.createFaultAsXmlBeans(true);
                        UnicastXmlMessage responseMessage = new UnicastXmlMessage(faultMessage, node
                                .getId(), soapAction + "Response");
                        RelatesToDocument relatesToDocument = RelatesToDocument.Factory.newInstance();
                        relatesToDocument.addNewRelatesTo().setStringValue(
                                requestMessage.getMessageID());
                        responseMessage.addHeader(relatesToDocument);
                        sendMessage(returnId, responseMessage);
                    }
                } catch (Throwable e1) {
                    log.error(e1);
                }

            }
        }
        return true;
        // TODO if intended recipeint is different from current node that means
        // that node is down. Respond to sender saying it was a error
    }

    /**
     * This method is invoked on applications when the underlying node is about
     * to forward the given message with the provided target to the specified
     * next hop. Applications can change the contents of the message, specify a
     * different nextHop (through re-routing), or completely terminate the
     * message.
     * 
     * @param message
     *                The message being sent, containing an internal message
     *                along with a destination key and nodeHandle next hop.
     * 
     * @return Whether or not to forward the message further
     */
    public boolean forward(RouteMessage arg0) {
        return true;
    }

    /**
     * This method is invoked to inform the application that the given node has
     * either joined or left the neighbor set of the local node, as the set
     * would be returned by the neighborSet call.
     * 
     * @param handle
     *                The handle that has joined/left
     * @param joined
     *                Whether the node has joined or left
     */
    public void update(rice.p2p.commonapi.NodeHandle arg0, boolean arg1) {
    }

    /**
     * Send a message to a give recipeint
     * 
     * @param recipient
     * @param message
     */
    public void sendMessage(Id recipient, UnicastXmlMessage message) {
        log.debug("Send message  from " + endpoint.getId() + " to " + recipient);
        if (message.getTo() == null) {
            ToDocument toDocument = ToDocument.Factory.newInstance();
            toDocument.addNewTo().setStringValue(recipient.toStringFull());
            message.addHeader(toDocument);
        }
        endpoint.route(recipient, message, null);
    }

    /**
     * Send a message to a give recipient and wait for a response. We do not
     * support time outs as of now.
     * 
     * @param recipient
     * @param message
     * @return
     * @throws HasthiException
     */
    public UnicastXmlMessage sendReciveMessage(Id recipient, UnicastXmlMessage message,
            int timeoutInseconds) throws InterruptedException {
        log.debug("Send message   from " + endpoint.getId() + " to " + recipient);
        ToDocument toDocument = ToDocument.Factory.newInstance();
        toDocument.addNewTo().setStringValue(recipient.toStringFull());
        message.addHeader(toDocument);
        endpoint.route(recipient, message, null);
        log.debug("Waiting for " + endpoint.getId() + " respond ");
        String reqeustMessageID = message.getMessageID();
        if (timeoutInseconds == -1) {
            return (UnicastXmlMessage) messageRendezvous.waitFor(reqeustMessageID);
        } else {
            return (UnicastXmlMessage) messageRendezvous
                    .waitFor(reqeustMessageID, timeoutInseconds);
        }
    }

    public UnicastXmlMessage sendReciveMessage(Id recipient, UnicastXmlMessage message)
            throws InterruptedException {
        return sendReciveMessage(recipient, message, -1);
    }

    /* (non-Javadoc)
     * @see hasthi.container.p2p.MulticastContianier#createCommunicationClient(hasthi.container.NodeAddress)
     */
    public CommunicationClient createCommunicationClient() {
        return new P2PCommunicationClient(this);
    }

    /* (non-Javadoc)
     * @see hasthi.container.p2p.MulticastContianier#send(hasthi.container.MessageContext)
     */
    public void send(MessageContext message) {
        UnicastXmlMessage requestMessage = new UnicastXmlMessage(message.getRequest(),
                node.getId(), message.getSoapAction());
        sendMessage(message.getAddress().getAddressAsObject(), requestMessage);
    }

    /* (non-Javadoc)
     * @see hasthi.container.p2p.MulticastContianier#sendReceive(hasthi.container.MessageContext)
     */
    public XmlObject sendReceive(MessageContext message) throws HasthiException {
        try {
            UnicastXmlMessage requestMessage = new UnicastXmlMessage(message.getRequest(), node
                    .getId(), message.getSoapAction());
            UnicastXmlMessage responseMessage = sendReciveMessage(message.getAddress()
                    .getAddressAsObject(), requestMessage);
            XmlObject responsebody = responseMessage.getBodyContent();
            if (SOAPUtils.isFaultMessage(responsebody)) {
                throw HasthiException.buildException(responsebody);
            } else {
                return responsebody;
            }
        } catch (InterruptedException e) {
            throw new HasthiException(e, FaultCode.ErrorAtClientAfterWsCall);
        }
    }

    public PastryNode getNode() {
        return node;
    }

    public class P2PCommunicationClient implements CommunicationClient {
        private P2PContainer container;

        public P2PCommunicationClient(P2PContainer container) {
            this.container = container;
        }

        public void send(MessageContext message) throws HasthiException {
            container.send(message);
        }

        public XmlObject sendReceive(MessageContext message) throws HasthiException {
            return container.sendReceive(message);
        }
    }

    /* (non-Javadoc)
     * @see hasthi.container.p2p.MulticastContianier#getContainierAddress()
     */
    public String getContainierAddress() {
        return getNode().getNodeId().toStringFull();
    }

    // public void findPeers(){
    // RoutingTable tr = node.getRoutingTable();
    // for(int i = 0;i<tr.numRows();i++){
    // RouteSet[] rows = tr.getRow(i);
    // for(RouteSet set:rows){
    // if(set != null && set.closestNode() != null){
    // System.out.println(set.closestNode().getId());
    // }
    // }
    // }
    // }
    
    
    
}
